// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu, jhkim

#pragma once

#include <experimental/filesystem>
#include <map>
#include <memory>
#include <string>
#include <tuple>
#include <vector>

#include "coin2/exchange/base/api_util/auth_key.h"
#include "coin2/exchange/base/feed/update.h"
#include "coin2/exchange/base/symbology/product.h"
#include "coin2/exchange/base/symbology/product_encyclopedia.h"

#include "coin2/strategy/linear_model/logic.h"
#include "coin2/strategy/linear_model/lm_config.pb.h"
#include "coin2/strategy/linear_model/halt_manager.h"
#include "coin2/strategy/linear_model/util_ban_vu.h"

#include "coin2/strategy/strategy.h"
#include "coin2/strategy/util/alarm.h"

#include "presto/quant/math/moving_average.h"

#include "presto/quant/feature/feature_manager.h"
#include "presto/quant/feature/linear_model/linear_model.h"  // LinearModel, LinearModelDelegate
#include "presto/quant/feature/linear_model/linear_model_delegate_proto.h"

namespace coin2::strategy::linear_model {

using ::coin2::exchange::base::feed::FeedUpdate;
using ::coin2::exchange::base::symbology::IProduct;
using ::coin2::strategy::IStrategy;
using ::presto::Symbol;
using ::presto::SymbolManager;
using ::presto::quant::feature::FeatureManager;
using ::presto::quant::feature::linear_model::LinearModel;
using ::presto::quant::feature::linear_model::LinearModelDelegateProto;

class BboFilter {
 public:
  bool OnBook(const FeedUpdate& upd) {
    std::tuple<double, double, double, double> bbo_tuple = {upd.book().Ask0()->price,
                                                            upd.book().Ask0()->qty,
                                                            upd.book().Bid0()->price,
                                                            upd.book().Bid0()->qty};
    bool sampled = bbo_tuple != prev_bbo_tuple_;
    prev_bbo_tuple_ = bbo_tuple;
    return sampled;
  }

 private:
  std::tuple<double, double, double, double> prev_bbo_tuple_{};
};

const Symbol* GetSymbolInline(
    const SymbolManager* symbol_manager,
    const std::string& ex,
    const std::string& norm);


struct SymbolWarmup {
  void UpdateFeed(
        const Symbol* symbol,
        const FeedMessage& feed_msg,
        int64_t disconn_thres_ns) {
    if (feed_msg.has_trade()) return;
    // if reconnected
    bool reconnected = (
        feed_msg.fetched_time() - last_ft_ > disconn_thres_ns);
    bool ft_et_shit = (
        feed_msg.has_exchange_time() &&
        feed_msg.exchange_time() + 10e9 < feed_msg.fetched_time());
    // block 30 seconds
    if (reconnected) {
      LOG_EVERY_N(ERROR, 50)
          << "Feed reconnected? ban trading temporarily: "
          << symbol->symbol_string() << " "
          << feed_msg.DebugString();

      block_until_time_ = feed_msg.fetched_time() + 10'000'000'000L;
      LOG(ERROR) << symbol->symbol_string() << " is BLOCKED since "
          << feed_msg.fetched_time() << " until "
          << block_until_time_;
    } else if (ft_et_shit) {
      LOG_EVERY_N(ERROR, 50)
          << "FT - ET > 10e9, ban trading temporarily: "
          << symbol->symbol_string() << " "
          << feed_msg.DebugString();

      block_until_time_ = feed_msg.fetched_time() + 10'000'000'000L;
      LOG(ERROR) << symbol->symbol_string() << " is BLOCKED since "
          << feed_msg.fetched_time() << " until "
          << block_until_time_;
    }
    last_ft_ = feed_msg.fetched_time();
    last_et_ = std::max(
        last_et_,
        feed_msg.has_exchange_time() ?
        feed_msg.exchange_time() : feed_msg.fetched_time());
  }

  bool IsOutdated(int64_t timestamp) const {
    return timestamp - last_et_ > 30'000'000'000L;
  }

  bool IsBlocked(const FeedMessage& feed_msg) {
    return (feed_msg.fetched_time() < block_until_time_);
  }

  int64_t last_ft_ = 0L;
  int64_t last_et_ = 0L;
  int64_t block_until_time_ = 0L;
};


struct StrategyWarmupManager {
  void UpdateFeed(
        const Symbol* symbol,
        const FeedMessage& feed_msg,
        int64_t disconn_thres_ns) {
    symbol_warmups_[symbol].UpdateFeed(symbol, feed_msg, disconn_thres_ns);
    bool blocked = symbol_warmups_[symbol].IsBlocked(feed_msg);
    if (blocked) {
      last_block_time_ = feed_msg.fetched_time();
      last_block_symbol_ = symbol;
    }
  }

  const Symbol* IsBlockedAny(const FeedMessage& feed_msg) {
    if (feed_msg.fetched_time() - last_block_time_ > 5'000'000'000L) {
      return nullptr;
    } else {
      return last_block_symbol_;
    }
  }

  const Symbol* IsOutdatedAny(int64_t timestamp) {
    if (timestamp - last_outdated_check_time_ > 5'000'000'000L) {
      bool outdated = false;
      for (const auto& [symbol, warmup] : symbol_warmups_) {
        if (warmup.IsOutdated(timestamp)) {
          LOG_EVERY_N(ERROR, 50) << symbol->symbol_string() << " is OUTDATED";
          if (!outdated_) last_outdated_time_ = timestamp;
          last_outdated_symbol_ = symbol;
          outdated = true;
        }
      }
      if (outdated_ && outdated) {
        reactivated_time_ = timestamp;
      }
      outdated_ = outdated;
      last_outdated_check_time_ = timestamp;
    }
    if (timestamp - reactivated_time_ < 120'000'000'000L) {
      LOG_EVERY_N(ERROR, 50) << "back from OUTDATED: Wait 120 sec after reactivation";
      // still outdated until reactivated
      return last_outdated_symbol_;
    }
    if (outdated_) {
      return last_outdated_symbol_;
    } else {
      return nullptr;
    }
  }

  int64_t DurationOutdated(int64_t timestamp) const {
    return timestamp - last_outdated_time_;
  }

  presto::detail::FastSymbolMap<SymbolWarmup> symbol_warmups_;
  int64_t last_block_time_ = 0L;
  const Symbol* last_block_symbol_ = nullptr;
  bool outdated_ = false;
  int64_t last_outdated_check_time_ = 0L;
  int64_t last_outdated_time_ = 0L;
  const Symbol* last_outdated_symbol_ = nullptr;
  int64_t reactivated_time_ = 0L;
};


struct SymbolHaltManager {
  void AddSymbol(const Symbol* symbol) {
    symbol_set_[symbol] = true;
  }

  void InitTimestamp(int64_t timestamp) {
    if (!init_) {
      for (const auto& [symbol, dummy] : symbol_set_) {
        (void)dummy;
        InitSymbol(symbol, timestamp);
      }
      init_ = true;
    }
  }

  void InitSymbol(const Symbol* symbol, int64_t start_ts) {
    if (symbol->symbol_string().find("WEEK", 0) != std::string::npos) {
      // expiry: 8am Friday UTC
      // week: effectively expires 1 hrs before expiry
      expiries_[symbol] = ::impl::Weekly(start_ts, 7);
    }
  }

  bool IsHalt(const Symbol* symbol, int64_t timestamp) {
    if (!init_) {
      InitTimestamp(timestamp);
    }
    return (
        expiries_.count(symbol) > 0 &&
        timestamp > expiries_.at(symbol));
  }

  bool init_ = false;
  presto::detail::FastSymbolMap<bool> symbol_set_;
  presto::detail::FastSymbolMap<int64_t> expiries_;
};


class FeatureFeedConverter {
 public:
  FeatureFeedConverter(
      const SymbolManager* symbol_manager,
      ::coin2::exchange::base::symbology::ProductEncyclopedia* product_cache,
      bool register_symbols = false,
      std::optional<int64_t> tfeed_etft_guard_ns = 2e9);

  void set_qty_multiplier(double qty_multiplier) { this->qty_multiplier = qty_multiplier; }

  void RegisterSymbols();

  virtual ~FeatureFeedConverter() = default;

  const std::string& GetCoin1ExchangeNameFromEm(
      coin::proto::ExchangeType exchange,
      coin::proto::MarketType market) const {
    return em_map_.at(std::make_tuple(exchange, market));
  }

  const std::string& GetCoin1ExchangeName(const MarketExchangeApi& mea) const {
    return GetCoin1ExchangeNameFromEm(mea.exchange, mea.market);
  }

  bool HasMea(const MarketExchangeApi& mea) const {
    return em_map_.count(std::make_tuple(mea.exchange, mea.market)) > 0;
  }

  void onTradeFeed(const FeedUpdate& upd);
  void onBookFeed(const FeedUpdate& upd);
  void onStatusFeed(const FeedUpdate& upd);
  void onFeed(const Symbol* symbol, const FeedUpdate& upd);
  virtual const Symbol* GetSymbol(const FeedUpdate& upd);

 public:
  bool feed_converted = false;
  ::fastfeature::FeedMessage last_feed_msg{};
  const SymbolManager* symbol_manager;
  std::optional<int64_t> tfeed_etft_guard_ns;
  double qty_multiplier = 1;
  std::optional<int64_t> before_feature_calc_time;

 protected:
  SymbolHaltManager symbol_halt_;

 private:
  std::map<
      std::tuple<coin::proto::ExchangeType, coin::proto::MarketType>,
      std::string> em_map_;
  ::coin2::exchange::base::symbology::ProductEncyclopedia* product_cache_;
};

class ProductLmStrategy : public IStrategy {
 public:
  ProductLmStrategy(
      LinearModel* linear_model,
      int64_t* lm_feed_id,
      const int64_t* global_feed_id,
      LinearModelDelegateProto* delegate,
      const MarketExchangeApi& mea,
      std::string symbol_str,
      const LmProductConfig& lm_product_config,
      ::coin2::exchange::base::symbology::ProductEncyclopedia* product_cache,
      int64_t timestamp_override,
      bool simulation,
      const VolumeInfo& volume_info,
      const std::vector<presto::Mep>& clear_meps,
      std::shared_ptr<::fastfeature::KlineInfoProto> kline_info);

  void Init(::coin2::app::Driver* driver) override {
    const auto& order_configs = driver->config().order().exchanges();
    CHECK_EQ(order_configs.size(), 1);
    const auto& order_config = order_configs.begin()->second;
    MarketExchangeApi mea = MarketExchangeApi::FromString(order_config.mea());
    const auto& key_filepath = order_config.connection_config().key_filepath();

    coin::proto::AccountRequestProto acc_request;
    acc_request.set_market_type(MarketType_Name(mea.market));
    acc_request.set_exchange(ExchangeType_Name(mea.exchange));
    if (!key_filepath.empty()) {
      AuthKey auth_key = AuthKey::FromFile(key_filepath);
      acc_request.set_owner(auth_key.owner());
    } else {
      acc_request.set_owner("sim");
    }
    linear_model_logic_->SetStartTimestamp(timestamp_override_);
    // TODO: must CHECK strat_logger_?
    if (strat_logger_) {
      linear_model_logic_->Init(
          driver,
          strat_logger_,
          acc_request);
    }
  }

  void RegisterAlarm(util::AlarmManager* alarm) { alarm_ = alarm; }

  void onTradeFeed(const FeedUpdate& upd) override;
  void onBookFeed(const FeedUpdate& upd) override;
  void onStatusFeed(const FeedUpdate& upd);
  void onFeatureFeed(
      const FeedUpdate& upd,
      const ::fastfeature::FeedMessage* feed_msg,
      bool simulation);
  LinearModelLogic* GetLinearModelLogicMutable() { return linear_model_logic_.get(); }
  const IProduct* GetTargetProduct() { return target_product_.get(); }
  const Symbol* ResolveSymbol(const MarketExchangeApi& mea, const std::string& symbol_str) const {
    return GetSymbolInline(
        converter->symbol_manager,
        converter->GetCoin1ExchangeName(mea),
        symbol_str);
  }

  const SymbolManager* GetSymbolManager() const {
    return converter->symbol_manager;
  }

  const FeatureManager* GetFeatureManager() const {
    return linear_model_->GetFeatureManager();
  }

  void UpdateMidp(StrategyReporter2* strat_reporter) {
    if (!simulation_) {
      linear_model_logic_->UpdateMidp(strat_reporter, GetTargetProduct());
    }
  }

  void OnFill(
      int64_t timestamp,
      coin::proto::TradeSide trade_side,
      double fill_price,
      double fill_qty) {
    if (linear_model_logic_) {
      linear_model_logic_->OnFill(timestamp, trade_side, fill_price, fill_qty);
    }
  }

  void UpdateAccountInfo(const coin2::exchange::base::order::OrderUpdate& upd) {
    if (linear_model_logic_) {
      linear_model_logic_->onAccountInfo(upd);
      SPDLOG_INFO_EVERY_NS(
          200'000'000'000L,
          upd.timestamp(),
          "onAccountInfo({}) processed",
          upd.is_account_info());
    }
  }

  void UpdateAccountOrder(const coin2::exchange::base::order::OrderUpdate& upd) {
    if (linear_model_logic_) {
      linear_model_logic_->onAccountOrder(upd);
      SPDLOG_INFO_EVERY_NS(
          200'000'000'000L,
          upd.timestamp(),
          "onAccountOrder processed: {}",
          upd.product_order_info().event().ShortDebugString());
    }
  }

  void set_strategy_logger(StrategyLogger* strat_logger) { strat_logger_ = strat_logger; }

  std::unique_ptr<FeatureFeedConverter> converter;

 private:
  bool is_focus_;
  LinearModel* linear_model_;
  int64_t* lm_feed_id_;
  const int64_t* global_feed_id_;
  LinearModelDelegateProto* delegate_;
  const bool simulation_;
  const bool ftet_guard_;
  const bool force_try_clear_;
  HaltManager halt_manager_;
  std::unique_ptr<LinearModelLogic> linear_model_logic_;
  const Symbol* target_;
  std::unique_ptr<IProduct> target_product_;
  StrategyLogger* strat_logger_;
  StrategyWarmupManager warmup_;
  util::AlarmManager* alarm_ = nullptr;
  int64_t lastlogt_blocked_ = 0L;
  int64_t lastlogt_outdated_ = 0L;
  int64_t lastlogt_alarm_ = 0L;
  int64_t lastlogt_warning_ = 0L;
  int64_t timestamp_override_ = 0L;

  static constexpr int64_t kDefaultClearTimeout_ = 300L * 1e9;
};

}  // namespace coin2::strategy::linear_model
